package tl_client

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"sync/atomic"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"

	// ips

	pb_stream "io.plus/test_loss/pkg/stream"
)

type ClientArgs struct {
	Target string
	// second
	Duration    int
	JobDir      string
	JobID       string
	JobFullPath string
	IsDsync     bool
	IsFsync     bool
	UseDB       bool
}

var GlbClientArgs ClientArgs

func sendWorker(conn *grpc.ClientConn, done chan struct{}, stopmain chan struct{}) {
	canClose := atomic.Bool{}
	canClose.Store(true)
	needstop := atomic.Bool{}
	c := pb_stream.NewStreamServiceClient(conn)

	// 创建一个流
	stream, err := c.StreamCounts(context.Background())
	if err != nil {
		log.Fatalf("could not stream: %v", err)
	}

	go func() {
		err := os.RemoveAll(GlbClientArgs.JobFullPath)
		if err != nil {
			log.Fatalf("remove job dir err: %v\n", err)
		}

		err = os.MkdirAll(GlbClientArgs.JobFullPath, 0o777)
		if err != nil {
			log.Fatalf("create job dir err: %v\n", err)
		}

		defer func() {
			canClose.Store(true)
		}()
		// 发送数据流

		job_info_path := GlbClientArgs.JobFullPath + "_job_info"
		// 将 GlbClientArgs 序列化为 json
		jobInfoJSON, err := json.MarshalIndent(GlbClientArgs, "", "  ")
		if err != nil {
			log.Fatalf("Failed to marshal job args: %v", err)
		}

		// jobInfoJSON 转化为 string
		jobInfoStr := string(jobInfoJSON)

		WrtieFile(job_info_path, jobInfoStr)

		for i := int64(1); ; i++ {
			// 判断 stream grpc.BidiStreamingClient  是否关闭
			if !canClose.CompareAndSwap(true, false) {
				break
			}
			if needstop.CompareAndSwap(true, false) {
				break
			}

			if err := stream.Send(&pb_stream.CountRequest{Count: i, Jobid: GlbClientArgs.JobID}); err != nil {
				log.Fatalf("Failed to send a count: %v", err)
			}
			resp, err := stream.Recv()

			if err == io.EOF {
				break
			} else if err != nil {
				log.Fatalf("Failed to receive a count: %v", err)
			}
			go func() {
				log.Printf("Received count: %v", resp.GetResult())
			}()
			id_str := fmt.Sprintf("%d", i)
			abs_file_path := filepath.Join(GlbClientArgs.JobFullPath, id_str)
			if GlbClientArgs.UseDB {
				// TODO: 使用数据库
				SaveJobInfoToDB(GlbClientArgs.JobID, i)
			} else {
				DioWrtieFile(abs_file_path, id_str)
			}
			canClose.Store(true)
		}
	}()

	<-done
	needstop.Store(true)

forLoop:
	for {
		if canClose.CompareAndSwap(true, false) {
			if err := stream.CloseSend(); err != nil {
				if err != io.EOF {
					log.Fatalf("stream.CloseAndRecv() failed: %v", err)
				}
			}
			log.Printf("stream closed normally")
			break forLoop
		}
		log.Printf("wait for stream closeing...")
		time.Sleep(100 * time.Millisecond)
	}

	stopmain <- struct{}{}
}

// serverCmd represents the server command
func Run() {
	cli_args := GlbClientArgs
	// 连接到gRPC服务器
	conn, err := grpc.NewClient(cli_args.Target,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close()
	done := make(chan struct{})
	stopmain := make(chan struct{})

	go func() {
		sendWorker(conn, done, stopmain)
	}()

	if cli_args.Duration <= 0 {
		log.Printf("client nerver timeout")
		for {
			time.Sleep(time.Second * 3600 * 24 * 365) // 休眠一年
		}
	}
	time.AfterFunc(time.Duration(cli_args.Duration)*time.Second, func() {
		log.Printf("Time is up, signaling goroutine to exit, can be set with --duration 0, dura is %v s", cli_args.Duration)
		close(done) // 关闭通道，发送退出信号
	})
	<-stopmain
}
